package com.zjh.reactive;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;

public class FluxDemo {
    public static void main1(String[] args) {
        List<Integer> list = List.of();

        Flux.range(1, 10)
                .doOnNext(System.out::println)
                .subscribe();
        Flux.from(new Publisher<Integer>() {

            @Override
            public void subscribe(Subscriber<? super Integer> subscriber) {
                subscriber.onSubscribe(new Subscription() {
                                           @Override
                                           public void request(long l) {
                                               for (int i = 0; i < 10; i++) {
                                                   subscriber.onNext(1);

                                               }
                                               subscriber.onComplete();
                                           }

                                           @Override
                                           public void cancel() {

                                           }
                                       }
                );
            }
        }).subscribe(
                v -> System.out.println("收到消息：" + v),
                e -> System.out.println("出错：" + e),
                () -> System.out.println("完成")
        );

    }

    public static void main2(String[] args) {
        Flux.defer(() -> new Publisher<String>() {
            @Override
            public void subscribe(Subscriber<? super String> subscriber) {
                subscriber.onSubscribe(new Subscription() {
                    @Override
                    public void request(long l) {
                        subscriber.onNext("nihao");
                    }

                    @Override
                    public void cancel() {

                    }
                });
            }
        }).subscribe(v -> System.out.println("收到消息：" + v),
                e -> System.out.println("出错：" + e),
                () -> System.out.println("完成"));
    }

    public static void main3(String[] args) {
        Flux.range(1, 5)
                .index()
                .subscribe(System.out::println);
        Flux.range(1, 5)
                .timestamp()

                .subscribe(System.out::println);
        /*
            [0,1]
            [1,2]
            [2,3]
            [3,4]
            [4,5]
            [1762081788571,1]
            [1762081788573,2]
            [1762081788573,3]
            [1762081788573,4]
            [1762081788573,5]
         */

    }

    public static void main6(String[] args) {
//        Flux.range(1,5)
//                .scan(0,(num1,num2) ->{
//                    System.out.println("num1:" + num1);
//                    System.out.println("num2:" + num2);
//                    return num1 + num2;
//                })
//                .map(e->e*2)
//                .subscribe(System.out::println);
//        List<Integer> list = new ArrayList<>();
//        Flux<int[]> sca=Flux.just(new int[]{},new int[]{});
//        Flux<int[]> scan = Flux.range(1, 8)
////                .doOnNext(System.out::println)
//                .scan(new int[5], (arr, e) -> {
//                    arr[e%5]=e;
//                    return arr;
//                });
        int arrLength = 5;
        Flux.just(1,2,3,4,5,6)
                .index()
                .scan(new int[arrLength],(arr,entry) ->{
                    arr[(int) (entry.getT1() % arrLength)] = entry.getT2();
                    return arr;
                })
//                .skip(arrLength)
//              // 当窗口数组背灌满之后，开始计算平均值，因此要跳过没有灌满的情况
//              scan 的第一个发射是 初始数组 [0,0,0,0,0]！
                .doOnNext(s->{
                    for (int i=0;i<arrLength;i++){
                        System.out.println(s[i]);
                    }
                })
                .map(array -> Arrays.stream(array).sum() * 1.0 /arrLength)
                .subscribe(System.out::println);


    }

    public static void main(String[] args) throws InterruptedException {
//        Flux.just(1, 2)
//                .flatMap(e -> Flux.just(e * 2));
//
//        Flux<String> map = Flux.just("A", "B", "C")
//                .map(s -> s.toLowerCase());
//        Flux.just(Flux.range(1,2),Flux.range(3,4));

//        Flux.just(Flux.range(1,2),Flux.range(3,4))
//                .log()
//                .doOnNext(integerFlux -> {
//                    integerFlux.collectList().subscribe(System.out::println);
//                })
//                .map(integerFlux -> {
//                  return    integerFlux.collectList();
//                })
//                .log()
//                .subscribe(s->{
//                    s.subscribe(System.out::println);
//                });

//        Flux.just(Flux.range(1,2),Flux.range(3,4))
//                .log()
//                .doOnNext(integerFlux -> {
//                    integerFlux.collectList().subscribe(System.out::println);
//                })
//
//                .flatMap(integerFlux -> {
//                    return    integerFlux.collectList();
//                })
//                .log()
//                .subscribe(System.out::println);


//        Flux.range(1,4)
//                .map(s->s*2)
//                .log()
//                .subscribe(System.out::println);
////        Flux.range(1,4)
////                .flatMap(s->Flux.just(s*2))
////                .log()
////                .subscribe(System.out::println);
//        Random random = new Random();
//        Flux.just(Arrays.asList(1,2,3),Arrays.asList("a","b","c","d"),Arrays.asList(7,8,9))
////                .doOnNext(System.out::println)
//                .log()
//                .concatMap(item -> Flux.fromIterable(item)
//                        .doOnSubscribe(subscription -> {
//                            System.out.println("已经订阅");
//                        })// 我们增加一个延时，订阅后延时一段时间再发送
//                        .delayElements(Duration.ofMillis(random.nextInt(100) + 100))
//                ).subscribe(System.out::println);
//
//        Thread.sleep(10*1000);
        Flux<String> stringFlux = Mono.just("hellozjh").flatMapMany(e -> {
            String[] split = e.split("");
            System.out.println(split);
            return Flux.fromArray(split);
        });
        stringFlux.subscribe(System.out::println);
//        Mono<String> mono = Mono.just("abc");
//
//        Flux<Character> flux = mono.flatMapMany(str -> {
//            List<Character> list = str.chars()
//                    .mapToObj(c -> (char) c)
//                    .collect(Collectors.toList());
//            return Flux.fromIterable(list);
//        });
//
//        flux.log()
//                .subscribe(System.out::println);
//        Flux.just("a","b","c");

    }
}
